package com.act.task;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.CarbonContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.util.CommonUtils;
import com.act.util.JdbcUtil;
import com.act.util.SystemConfig;
import com.act.util.URLUtil;

public class DomainTask {

	private static final Log log = LogFactory.getLog(DomainTask.class);
		
	@SuppressWarnings("serial")
	public static void main(String[] args){
		try{
			String storeLocation = SystemConfig.getProperty("storeLocation");
			String carbonDB = SystemConfig.getProperty("carbonDB");
			
			SparkConf conf = new SparkConf();
			conf.setAppName("DomainTask");
			SparkContext sc = new SparkContext(conf);
			CarbonContext cc = new CarbonContext(sc,storeLocation);
			
			String dayStr = "";
			if(args.length > 0){
				//时间格式:20171026
				dayStr = args[0];
			}
			log.info("传入的时间："+dayStr);
			String tableName = carbonDB+".t_detector_check_fraud_"+getExecDate(dayStr);
			DataFrame domainFrame = cc.sql("select type,isfraud,fraudtype,subfraudtype,country,province,area,host,url,urlparam,ip,state,control,isforeign,isweb,operator,md5,appname,sitetype,sitetypename,checkmonth,checkdate,checkhour,checktime,time,logtime,uuid,modifytime from "+tableName+" where type= '0'");

			JavaRDD<Map<String,Object>> rdd = domainFrame.toJavaRDD().mapPartitions(new FlatMapFunction<Iterator<Row>,Map<String,Object>>() {

				@Override
				public Iterable<Map<String,Object>> call(Iterator<Row> it){
					List<Map<String,Object>> list = new ArrayList<Map<String,Object>>();
					String host = "";
					while(it.hasNext()){
						try{
							Row row = it.next();
							host = row.getAs("host");
							String topDomain = URLUtil.getDomainName("http://" + host);
							if(StringUtils.isNotBlank(topDomain)){
								Map<String,Object> map = new HashMap<String,Object>();
								map.put("id", 0);
								map.put("type", row.getAs("type"));
								map.put("isfraud", row.getAs("isfraud"));
								map.put("fraudtype", row.getAs("fraudtype"));
								map.put("subfraudtype", row.getAs("subfraudtype"));
								map.put("country", row.getAs("country"));
								map.put("province", row.getAs("province"));
								map.put("area", row.getAs("area"));
								map.put("host",host);
								map.put("url", row.getAs("url"));
								map.put("urlparam", row.getAs("urlparam"));
								map.put("ip", row.getAs("ip"));
								map.put("state", row.getAs("state"));
								map.put("control", row.getAs("control"));
								map.put("isforeign", row.getAs("isforeign"));
								map.put("isweb", row.getAs("isweb"));
								map.put("operator", row.getAs("operator"));
								map.put("md5", row.getAs("md5"));
								map.put("appname", row.getAs("appname"));
								map.put("sitetype", row.getAs("sitetype"));
								map.put("sitetypename", row.getAs("sitetypename"));
								map.put("checkmonth", row.getAs("checkmonth"));
								map.put("checkdate", row.getAs("checkdate"));
								map.put("checkhour", row.getAs("checkhour"));
								map.put("checktime", row.getAs("checktime"));
								map.put("time", row.getAs("time"));
								map.put("logtime", row.getAs("logtime"));
								map.put("uuid",row.getAs("uuid"));
								map.put("modifytime", row.getAs("modifytime"));
								map.put("checkdate", row.getAs("checkdate"));
								map.put("isdail", "");
								map.put("isdomain","");
								map.put("location", "");
								map.put("topdomain",topDomain);
								list.add(map);
							}
						}catch(Exception e){
							log.info("当前解析出错的域名为："+host);
							if(log.isErrorEnabled()){
								log.error(e.getMessage(),e);
							}
						}
					}
					return list;
				}
			});
			
			insertData(rdd);
			
			sc.stop();
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
	}
	
	@SuppressWarnings("serial")
	private static void insertData(JavaRDD<Map<String,Object>> rdd) {
		
		String columns = "id,type,country,province,area,host,url,urlparam,ip,state,control,isforeign,isweb,operator,md5,appname,sitetype,sitetypename,checkmonth,checkdate,checkhour,checktime,time,logtime,uuid,modifytime,isdail,isdomain,location,isfraud,fraudtype,subfraudtype,topdomain";
		String[] columnsSplit = columns.split(",");
		if(!rdd.isEmpty()){
//			String sql = "truncate table ods_idc_t_detector_check_fraud";
			String sql = "truncate table t_detector_check_fraud_temp";
			boolean flag = execSql(sql);
			if(flag){
				rdd.foreachPartition(new VoidFunction<Iterator<Map<String,Object>>>() {

					@Override
					public void call(Iterator<Map<String,Object>> it) {
						Connection conn = null;
						PreparedStatement ps = null;
						try {
							log.info("====当前分区插入数据开始====");
							conn = JdbcUtil.getConnection();
							
//							String sql = "insert into ods_idc_t_detector_check_fraud("+columns+") values(CHECK_FRAUD_SE.nextval,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
							String sql = "insert into t_detector_check_fraud_temp("+columns+") values(CHECK_FRAUD_SE.nextval,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
							conn.setAutoCommit(false);
							ps = conn.prepareStatement(sql);
							int n = 0;
							while (it.hasNext()) {
								Map<String,Object> map = it.next();
								for (int i = 0; i < map.size()-1; i++) {
									ps.setObject(i+1, map.get(columnsSplit[i+1]));
								}
								ps.addBatch();
								if (++n % 10000 == 0) {
									log.info("当前分区插入的数据量："+n);
									ps.executeBatch();
									conn.commit();
								}
							}
							ps.executeBatch();
							conn.commit();
							log.info("当前分区插入的数据总量："+n);
							log.info("====当前分区插入数据结束====");
						} catch (Exception e) {
							if (log.isErrorEnabled()) {
								log.error(e.getMessage(), e);
							}
						} finally {
							JdbcUtil.close(ps, conn);
						}
					}
				});
			}
		}
	}
	
	public static String getExecDate(String dayStr) {
		if (StringUtils.isBlank(dayStr)) {
			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
			dayStr = sdf.format(new Date());
		}
		return dayStr;
	}
	
	public static void createSeqIfNotExists(Connection conn){
		try{
			String sequenceName = "CHECK_FRAUD_SE";
			CommonUtils.getSequenceName(sequenceName,conn);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
	}
	
	public static boolean execSql(String sql){
    	boolean flag = true;
    	Connection conn = null;
    	PreparedStatement ps = null;
    	try{
    		conn = JdbcUtil.getConnection();
    		createSeqIfNotExists(conn);
    		conn.setAutoCommit(false);
    		ps = conn.prepareStatement(sql);
          	ps.executeQuery();
            conn.commit();
          	flag = true;
          }
          catch(SQLException e){
         	 flag = false;
         	 log.error("jdbc batch commit sql:"+sql+" error" , e);
              try {
                  conn.rollback();
              } catch (SQLException e1) {
                  log.error("jdbc rollback error" , e1);
              }
          }finally{
        	  JdbcUtil.close(ps,conn);
          }
    	return flag ;
    }
}
